package com.raven.play.stream.test.scaladsl.tutorial

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.javadsl.GraphDSL.builder
import akka.stream.scaladsl.{Broadcast, Flow, GraphDSL, Merge, RunnableGraph, Sink, Source, Zip}
import akka.stream.{ActorMaterializer, ClosedShape}

import scala.concurrent.duration.DurationInt

/**
 * Non-linear components:
 * - fan-out
 * - fan-in
 *
 * Fan-out components:
 * - Broadcast
 * - Balance
 *
 * Fan-in components:
 * - Zip/ZipWith
 * - Merge
 * - Concat
 */
object $09GraphBasics extends App {

  implicit val system = ActorSystem("GraphBasics")
  implicit val materializer = ActorMaterializer()

  val input = Source(1 to 1000)
  val incrementer = Flow[Int].map(x => x + 1) // hard computation
  val multiplier = Flow[Int].map(x => x * 10) // hard computation
  val output = Sink.foreach[(Int, Int)](print)

  // step 1 - setting up the fundamentals for the graph
  val graph = RunnableGraph.fromGraph(
    GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
      import GraphDSL.Implicits._ // brings some nice operators into scope

      // step 2 - add the necessary components of this graph
      val broadcast = builder.add(Broadcast[Int](2)) // fan-out operator
      val zip = builder.add(Zip[Int, Int]) // fan-in operator

      // step 3 - tying up the components
      input ~> broadcast
      broadcast.out(0) ~> incrementer ~> zip.in0
      broadcast.out(1) ~> multiplier ~> zip.in1

      zip.out ~> output

      // step 4 - return a closed shape
      ClosedShape // FREEZE THE BUILDER'S SHAPE
      // shape
    } //graph
  ) // runnable graph


  val firstSink = Sink.foreach[Int](x => println(s"First sink: $x"))
  val secondSink = Sink.foreach[Int](x => println(s"Second sink: $x"))

  val sourceToTwoSinkGraph = RunnableGraph.fromGraph(
    GraphDSL.create(){implicit builder: GraphDSL.Builder[NotUsed] =>
      import GraphDSL.Implicits._

      val broadcast = builder.add(Broadcast[Int](2))

//      input ~> broadcast
//      broadcast.out(0) ~> firstSink
//      broadcast.out(1) ~> secondSink

      input ~> broadcast ~> firstSink
               broadcast ~> secondSink

      ClosedShape
    }
  )

  val fastSource = input.throttle(5, 1.second)
  val slowSource = input.throttle(2, 1.second)

  val sink1 = Sink.fold[Int, Int](0)((count, element) => {
    println(s"Sink 1 got elements: $count")
    count + 1
  })
  val sink2 = Sink.fold[Int, Int](0)((count, element) => {
    println(s"Sink 2 got elements: $count")
    count + 1
  })

  val balanceGraph = RunnableGraph.fromGraph(
    GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
      import GraphDSL.Implicits._

      val merge = builder.add(Merge[Int](2))
      val broadcast = builder.add(Broadcast[Int](2))

      //      input ~> broadcast
      //      broadcast.out(0) ~> firstSink
      //      broadcast.out(1) ~> secondSink

      fastSource ~> merge ~> broadcast ~> sink1
      slowSource ~> merge
      broadcast ~> sink2

      ClosedShape
    }
  )

  balanceGraph.run() // run the graph and materialize it
}
